﻿using System;
using System.Collections.Generic;
using System.Text;
using System.Net.Sockets;
using System.IO;
using MapReduce.Comm.IO;
using System.Threading;
using MapReduce.Utils; 

namespace MapReduce.Comm
{


    public class JobRequest : BaseComm
    {
        public const byte SUCCESS = 1;
        public const byte ERROR = 0;

        public delegate void Callback(byte pStatus, byte [][] pData);

        private String mConnectTo;
        private ushort mPort;
        private const int CONNECTION_TIMEOUT = 50;

        public JobRequest(String pConnectTo, ushort pPort)
        {
            mConnectTo = pConnectTo;
            mPort = pPort;
        }

        public void ValidateConfig(String pConfigXmlPath, Callback pCallback)
        {
            String configXml = File.ReadAllText(pConfigXmlPath);

            byte[][] dataArr2D = { U.GetBytes(configXml) };
            Send("MapReduce.Services.CheckConfigService", dataArr2D, pCallback);
        }


        public void Send(String pClassName, byte[][] pData, Callback pCallback)
        {
 

            TcpClient tcpclnt = new TcpClient();
            tcpclnt.SendTimeout = CONNECTION_TIMEOUT;


            try
            {

                tcpclnt.Connect(mConnectTo, mPort); // use the ipaddress as in the server program


                this.IOStream = new StreamIOStream(tcpclnt.GetStream());

                U.Log("Send " + mConnectTo + ":" + mPort + "/" + pClassName);


                //Send class name
                this.SendData(U.GetBytes(pClassName));

                //Send if the response is expected
                if (pCallback == null)
                {
                    this.SendData(U.GetBytes("0"));
                }
                else
                {
                    this.SendData(U.GetBytes(Slave.Address + ":" + Slave.Port + ":" + JobCallbackPool.Lock(mConnectTo, mPort, pCallback)));
                }


                //Send data length
                this.SendData(BitConverter.GetBytes((ushort)pData.Length));

                //Send Data
                for (int i = 0; i < pData.Length; i++)
                {
                    this.SendData(pData[i]);
                }

                //Log
                Logger.Log(mConnectTo, pClassName, "Data Length:" + pData.Length);

            }
            catch (Exception ex)
            {
                Logger.LogError(mConnectTo, pClassName, ex);
                pCallback(JobRequest.ERROR, new byte[][] { U.GetBytes(mConnectTo + ":" + mPort + " " + ex.Message) });
            }

            finally 
            {
                this.Close();
                if (tcpclnt!=null) tcpclnt.Close();
            }
        }
    }
}
